package com.k.activemq.mqtt.demo1;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

/**
 * 如果使用 ActiveMQ 需要修改配置支持MQTT
 *  D:\apache\apache-activemq\apache-activemq-5.14.5\conf
 *  case:pass
 *
 * @author kivil
 */
public class MqttPublisher1 {
    public static void main(String[] args) {
        try{
            String url = "tcp://127.0.0.1:1883";
            String clientId = "client-publish-101";
            String username = "admin";
            String password="admin";
            MqttClient client = new MqttClient(url,clientId,new MemoryPersistence());
            client.setCallback(new MqttCallback() {
                @Override
                public void connectionLost(Throwable throwable) {
                    System.out.println(clientId+"与服务器断开连接");
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                    System.out.println(client.getClientId()+"发布消息成功！");

                }
            });
            MqttConnectOptions options = new MqttConnectOptions();
            //是否清空session，设置false表示服务器会保留客户端的连接记录（订阅主题，qos）,客户端重连之后能获取到服务器在客户端断开连接期间推送的消息
            //设置为true表示每次连接服务器都是以新的身份
            options.setCleanSession(true);
            //设置连接用户名
            options.setUserName(username);
            //设置连接密码
            options.setPassword(password.toCharArray());
            //设置超时时间，单位为秒
            options.setConnectionTimeout(100);
            //设置心跳时间 单位为秒，表示服务器每隔 1.5*20秒的时间向客户端发送心跳判断客户端是否在线
            options.setKeepAliveInterval(20);
            //设置遗嘱消息的话题，若客户端和服务器之间的连接意外断开，服务器将发布客户端的遗嘱信息
            options.setWill("willTopic",(clientId + "与服务器断开连接").getBytes(),0,false);
            client.connect(options);

            // publish
            String sendMsg = "{\"id\":\"101\"}";
            send(client,1,false,"mqtt-topic-101/a",sendMsg);
            client.disconnect();
            System.exit(0);
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * 在 MQTT 协议中，从 Broker 到 Subscriber 这段消息传递的实际 QoS 等于：Publisher 发布消息时指定的 QoS 等级和 Subscriber 在订阅时与 Broker 协商的 QoS 等级，这两个 QoS 等级中的最小那一个。
     * @param client
     * @param qos
     * @param retained
     * @param topic
     * @param message
     */
    public static void send(MqttClient client,int qos,boolean retained,String topic,String message){
        MqttMessage mqttMessage = new MqttMessage();
        // 0:消息最多传递一次，如果当时客户端不可用，则会丢失该消息
        // 1:消息传递至少 1 次。
        // 2:消息仅传送一次
        // 对于sender - broker 之间
        mqttMessage.setQos(qos);
        mqttMessage.setRetained(retained);
        mqttMessage.setPayload(message.getBytes());
        //主题的目的地，用于发布/订阅信息
        MqttTopic mqttTopic = client.getTopic(topic);
        //提供一种机制来跟踪消息的传递进度
        //用于在以非阻塞方式（在后台运行）执行发布是跟踪消息的传递进度
        MqttDeliveryToken token;
        try {
            //将指定消息发布到主题，但不等待消息传递完成，返回的token可用于跟踪消息的传递状态
            //一旦此方法干净地返回，消息就已被broker接受发布，当连接可用，将在后台完成消息传递。
            token = mqttTopic.publish(mqttMessage);
            token.waitForCompletion();
            boolean isComplete = token.isComplete();
            System.out.println("isComplete:"+isComplete);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
    public static void sendMessageWithRetry(MqttClient client, String topic, String payload, int maxRetries,int retryDelayMs) {
        int retryCount = 0;
        boolean isMessageSent = false;
        while (!isMessageSent && retryCount < maxRetries) {
            try {
                MqttMessage message = new MqttMessage(payload.getBytes());
                client.publish(topic, message);
                isMessageSent = true;
            } catch (MqttException e) {
                // 打印错误信息
                e.printStackTrace();
                // 等待一段时间后重试
                try {
                    Thread.sleep(retryDelayMs);
                } catch (InterruptedException ex) {
                    ex.printStackTrace();
                }
                retryCount++;
            }
        }
        if (!isMessageSent) {
            // 如果最终未能成功发送，可以在这里添加处理逻辑
            throw new RuntimeException("消息发送失败，已达到最大重试次数。");
        }
    }
}
